ROS2 自定义接口

自定义接口概念

1. 什么是接口

在 ROS 2 中，**接口（Interface）**是节点之间通信的"数据契约"。它定义了数据的结构和类型，确保不同节点之间能够正确理解和处理数据。

ROS 2 提供了三种主要接口类型：

消息（msg）：用于话题通信，单向、持续的数据流
服务（srv）：用于服务通信，请求-响应模式
动作（action）：用于动作通信，可反馈的长时间任务

2. 为什么需要自定义接口

ROS 2 内置了许多标准接口（如 std_msgs、sensor_msgs 等），但在实际开发中，我们常常需要定义自己的数据结构：

描述特定的业务数据（如学生信息、机器人状态等）
组合多种数据类型形成复合结构
定义特定服务请求和响应格式
实现跨语言、跨平台的数据交互

说明典型应用场景
自定义机器人传感器数据格式
定义特定的控制指令结构
封装复杂的状态信息
创建项目专用的服务接口

3. 接口包与功能包的关系

在 ROS 2 中，自定义接口通常放在独立的接口包中：

接口包：专门用于定义接口（msg、srv、action），不包含节点代码
功能包：包含实际的节点实现，依赖接口包

text

接口包 → 编译生成 → Python/C++ 代码 → 功能包调用

这种分离设计的好处：

复用性：一个接口包可以被多个功能包使用
解耦：接口定义与实现分离，便于维护
协作：团队成员可以各自开发功能包，共享接口定义

4. 语言无关性

ROS 2 接口的一个重要特性是语言无关性：

接口定义使用简单的文本格式（.msg、.srv 文件）
编译时自动生成各语言的代码（Python、C++ 等）
不同语言编写的节点可以无缝通信

说明关键点
无论使用 Python 还是 C++，只要引用相同的接口包，就能正确解析数据结构。

学习内容：自定义消息创建与使用

（一）创建接口包

首先进入工作空间的 src 目录，创建一个专门用于存放接口的包：

bash

cd ~/ros_code/lq_ws/src
ros2 pkg create --build-type ament_cmake student_interfaces

⚠️ 警告注意
接口包必须使用 ament_cmake 构建类型，不能使用 ament_python！

（二）创建 msg 和 srv 目录

进入接口包目录，创建存放消息和服务定义的文件夹：

bash

cd student_interfaces
mkdir msg
mkdir srv

此时目录结构如下：

text

student_interfaces/
├── CMakeLists.txt
├── package.xml
├── msg/
├── srv/
└── include/

（三）编写 Student.msg 自定义消息

在 msg 目录下创建 Student.msg 文件：

text

# msg/Student.msg
string name        # 学生姓名
int32 age           # 学生年龄
float64 height      # 学生身高
string grade        # 学生年级
bool is_graduated   # 是否毕业

消息定义语法说明：

字段类型	说明	示例
`int8/16/32/64`	有符号整数	`int32 age`
`uint8/16/32/64`	无符号整数	`uint8 count`
`float32/64`	浮点数	`float64 height`
`string`	字符串	`string name`
`bool`	布尔值	`bool active`
`数组`	使用 `[]`	`int32[] scores`

（四）编写 StudentInfo.srv 自定义服务

在 srv 目录下创建 StudentInfo.srv 文件：

text

# srv/StudentInfo.srv
# 请求部分
string student_id
---
# 响应部分
string name
int32 age
float64 height
string grade
bool found

说明服务定义格式
--- 上方为请求（Request）数据
--- 下方为响应（Response）数据

（五）修改 CMakeLists.txt

打开 CMakeLists.txt，进行以下修改：

1. 添加依赖

cmake

find_package(rosidl_default_generators REQUIRED)
find_package(rosidl_interface_packages REQUIRED)

2. 添加消息和服务定义

cmake

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Student.msg"
  "srv/StudentInfo.srv"
)

3. 添加导出依赖

cmake

ament_export_dependencies(rosidl_default_runtime)

完整的 CMakeLists.txt 示例：

cmake

cmake_minimum_required(VERSION 3.8)
project(student_interfaces)

if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
  add_compile_options(-Wall -Wextra -Wpedantic)
endif()

find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
find_package(rosidl_interface_packages REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Student.msg"
  "srv/StudentInfo.srv"
)

ament_export_dependencies(rosidl_default_runtime)

ament_package()

（六）修改 package.xml

打开 package.xml，添加必要的依赖声明：

xml

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>student_interfaces</name>
  <version>0.0.0</version>
  <description>自定义学生信息接口包</description>
  <maintainer email="user@todo.todo">user</maintainer>
  <license>TODO: License declaration</license>

  <buildtool_depend>ament_cmake</buildtool_depend>

  <build_depend>rosidl_default_generators</build_depend>
  <exec_depend>rosidl_default_runtime</exec_depend>
  <member_of_group>rosidl_interface_packages</member_of_group>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

说明关键依赖说明
rosidl_default_generators：用于编译时生成接口代码
rosidl_default_runtime：运行时依赖
rosidl_interface_packages：标记此包为接口包

（七）编译接口包

返回工作空间根目录，编译接口包：

bash

cd ~/ros_code/lq_ws
colcon build --packages-select student_interfaces
source install/setup.bash

验证接口包是否编译成功

bash

# 查看消息接口
ros2 interface show student_interfaces/msg/Student

# 查看服务接口
ros2 interface show student_interfaces/srv/StudentInfo

（八）创建使用接口的 Python 功能包

现在创建一个 Python 功能包来使用我们定义的接口：

bash

cd ~/ros_code/lq_ws/src
ros2 pkg create --build-type ament_python student_pkg --dependencies rclpy student_interfaces

（九）编写发布者节点

在功能包目录下创建 student_publisher.py：

python

import rclpy
from rclpy.node import Node
from student_interfaces.msg import Student

class StudentPublisher(Node):
    def __init__(self):
        super().__init__('student_publisher')
        self.publisher_ = self.create_publisher(Student, 'student_info', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.get_logger().info('学生信息发布者已启动')

    def timer_callback(self):
        msg = Student()
        msg.name = '张三'
        msg.age = 20
        msg.height = 175.5
        msg.grade = '三年级'
        msg.is_graduated = False

        self.publisher_.publish(msg)
        self.get_logger().info(f'发布学生信息: {msg.name}, {msg.age}岁, {msg.height}cm')

def main(args=None):
    rclpy.init(args=args)
    node = StudentPublisher()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

（十）编写订阅者节点

创建 student_subscriber.py：

python

import rclpy
from rclpy.node import Node
from student_interfaces.msg import Student

class StudentSubscriber(Node):
    def __init__(self):
        super().__init__('student_subscriber')
        self.subscription = self.create_subscription(
            Student,
            'student_info',
            self.listener_callback,
            10)
        self.get_logger().info('学生信息订阅者已启动')

    def listener_callback(self, msg):
        self.get_logger().info(
            f'收到学生信息:\n'
            f'  姓名: {msg.name}\n'
            f'  年龄: {msg.age}岁\n'
            f'  身高: {msg.height}cm\n'
            f'  年级: {msg.grade}\n'
            f'  是否毕业: {"是" if msg.is_graduated else "否"}'
        )

def main(args=None):
    rclpy.init(args=args)
    node = StudentSubscriber()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

（十一）编写服务端节点

创建 student_server.py：

python

import rclpy
from rclpy.node import Node
from student_interfaces.srv import StudentInfo

# 模拟数据库
STUDENT_DATABASE = {
    '001': {'name': '张三', 'age': 20, 'height': 175.5, 'grade': '三年级'},
    '002': {'name': '李四', 'age': 19, 'height': 168.0, 'grade': '二年级'},
    '003': {'name': '王五', 'age': 21, 'height': 180.0, 'grade': '四年级'},
}

class StudentServer(Node):
    def __init__(self):
        super().__init__('student_server')
        self.srv = self.create_service(StudentInfo, 'get_student_info', self.get_student_info_callback)
        self.get_logger().info('学生信息服务端已启动')

    def get_student_info_callback(self, request, response):
        student_id = request.student_id
        self.get_logger().info(f'收到查询请求，学号: {student_id}')

        if student_id in STUDENT_DATABASE:
            student = STUDENT_DATABASE[student_id]
            response.name = student['name']
            response.age = student['age']
            response.height = student['height']
            response.grade = student['grade']
            response.found = True
            self.get_logger().info(f'找到学生: {student["name"]}')
        else:
            response.found = False
            response.name = ''
            self.get_logger().info('未找到该学生')

        return response

def main(args=None):
    rclpy.init(args=args)
    node = StudentServer()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

（十二）编写客户端节点

创建 student_client.py：

python

import rclpy
from rclpy.node import Node
from student_interfaces.srv import StudentInfo

class StudentClient(Node):
    def __init__(self):
        super().__init__('student_client')
        self.client = self.create_client(StudentInfo, 'get_student_info')
        while not self.client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待服务端...')
        self.get_logger().info('学生信息客户端已连接')

    def send_request(self, student_id):
        request = StudentInfo.Request()
        request.student_id = student_id
        self.future = self.client.call_async(request)
        return self.future

def main(args=None):
    rclpy.init(args=args)
    client_node = StudentClient()

    # 查询学号为 001 的学生
    future = client_node.send_request('001')
    rclpy.spin_until_future_complete(client_node, future)

    if future.result() is not None:
        response = future.result()
        if response.found:
            client_node.get_logger().info(
                f'查询成功:\n'
                f'  姓名: {response.name}\n'
                f'  年龄: {response.age}岁\n'
                f'  身高: {response.height}cm\n'
                f'  年级: {response.grade}'
            )
        else:
            client_node.get_logger().info('未找到该学生')
    else:
        client_node.get_logger().error('服务调用失败')

    client_node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

（十三）修改 setup.py

打开功能包的 setup.py，添加入口点：

python

entry_points={
    'console_scripts': [
        'student_publisher = student_pkg.student_publisher:main',
        'student_subscriber = student_pkg.student_subscriber:main',
        'student_server = student_pkg.student_server:main',
        'student_client = student_pkg.student_client:main',
    ],
},

（十四）编译运行验证

1. 编译所有包

bash

cd ~/ros_code/lq_ws
colcon build
source install/setup.bash

2. 测试话题通信（自定义消息）

终端1 - 运行发布者：

bash

source install/setup.bash
ros2 run student_pkg student_publisher

终端2 - 运行订阅者：

bash

source install/setup.bash
ros2 run student_pkg student_subscriber

3. 测试服务通信（自定义服务）

终端1 - 运行服务端：

bash

source install/setup.bash
ros2 run student_pkg student_server

终端2 - 运行客户端：

bash

source install/setup.bash
ros2 run student_pkg student_client

4. 使用命令行工具验证

bash

# 查看话题列表
ros2 topic list

# 查看话题信息
ros2 topic info /student_info

# 手动发布消息
ros2 topic pub /student_info student_interfaces/msg/Student "{name: '测试', age: 22, height: 170.0, grade: '一年级', is_graduated: false}"

# 查看服务列表
ros2 service list

# 手动调用服务
ros2 service call /get_student_info student_interfaces/srv/StudentInfo "{student_id: '002'}"

接口类型速查表

常用基本类型

类型	Python 对应类型	说明
`bool`	`bool`	布尔值
`byte`	`int`	字节
`int8/16/32/64`	`int`	有符号整数
`uint8/16/32/64`	`int`	无符号整数
`float32/64`	`float`	浮点数
`string`	`str`	字符串

特殊类型

类型	说明
`类型[]`	动态数组，如 `int32[]`
`类型[N]`	固定长度数组，如 `float64[3]`
`Header`	ROS 标准头（时间戳、坐标系）
`其他包/消息`	嵌套消息，如 `geometry_msgs/msg/Pose`

说明最佳实践
接口包命名建议以 _interfaces 结尾
复杂的消息可以拆分为多个小消息组合使用
添加注释提高接口可读性
遵循 ROS 2 命名规范（小写字母、下划线分隔）

ROS2 自定义接口 ​

自定义接口概念 ​

1. 什么是接口 ​

2. 为什么需要自定义接口 ​

3. 接口包与功能包的关系 ​

4. 语言无关性 ​

学习内容：自定义消息创建与使用 ​

（一）创建接口包 ​

（二）创建 msg 和 srv 目录 ​

（三）编写 Student.msg 自定义消息 ​

（四）编写 StudentInfo.srv 自定义服务 ​

（五）修改 CMakeLists.txt ​

1. 添加依赖 ​

2. 添加消息和服务定义 ​

3. 添加导出依赖 ​

（六）修改 package.xml ​

（七）编译接口包 ​

验证接口包是否编译成功 ​

（八）创建使用接口的 Python 功能包 ​

（九）编写发布者节点 ​

（十）编写订阅者节点 ​

（十一）编写服务端节点 ​

（十二）编写客户端节点 ​

（十三）修改 setup.py ​

（十四）编译运行验证 ​

1. 编译所有包 ​

2. 测试话题通信（自定义消息） ​

3. 测试服务通信（自定义服务） ​

4. 使用命令行工具验证 ​

接口类型速查表 ​

常用基本类型 ​

特殊类型 ​

ROS2 自定义接口

自定义接口概念

1. 什么是接口

2. 为什么需要自定义接口

3. 接口包与功能包的关系

4. 语言无关性

学习内容：自定义消息创建与使用

（一）创建接口包

（二）创建 msg 和 srv 目录

（三）编写 Student.msg 自定义消息

（四）编写 StudentInfo.srv 自定义服务

（五）修改 CMakeLists.txt

1. 添加依赖

2. 添加消息和服务定义

3. 添加导出依赖

（六）修改 package.xml

（七）编译接口包

验证接口包是否编译成功

（八）创建使用接口的 Python 功能包

（九）编写发布者节点

（十）编写订阅者节点

（十一）编写服务端节点

（十二）编写客户端节点

（十三）修改 setup.py

（十四）编译运行验证

1. 编译所有包

2. 测试话题通信（自定义消息）

3. 测试服务通信（自定义服务）

4. 使用命令行工具验证

接口类型速查表

常用基本类型

特殊类型